package org.etnaframework.module.cron;

import org.etnaframework.module.base.spring.SpringContext;
import org.etnaframework.module.base.utils.CronExecutorUtils;
import org.etnaframework.module.base.utils.JacksonUtils;
import org.etnaframework.module.base.utils.ReflectUtils;
import org.etnaframework.module.base.utils.StackTraceUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
import org.springframework.data.redis.core.types.Expiration;

import java.lang.management.ManagementFactory;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 每个定时任务的具体内容
 *
 * @author jasyaf
 * @since 2024-04-06
 */
public class CronTask implements Runnable {

    static final Logger log = LoggerFactory.getLogger(CronTask.class);

    static final String REDIS_LOCK_PREFIX = "CronTaskLock::";

    static final long REDIS_LOCK_KEY1_EXPIRE_MS = 5 * 60 * 1000;

    static final long REDIS_LOCK_KEY2_EXPIRE_MS = 30 * 1000;

    /** 身份标识，标记是由哪个实例持有令牌 */
    final String pidAndHost;

    final ICronCustomizer customizer;

    /** 定时任务方法的位置（类名+方法名） */
    final String location;

    final String cron;

    final CronRunModeEnum runMode;

    final long timeoutMs;

    final Object bean;

    final Method method;

    long lastStartTime;

    long lastCostMs;

    volatile boolean running;

    volatile long nextTimeoutReportTimeMs;

    volatile boolean obtainLock;

    CronTask(ICronCustomizer customizer, String cron, CronRunModeEnum runMode, long timeoutMs, Object bean, Method method) {
        this.customizer = customizer;
        this.cron = cron;
        this.runMode = runMode;
        this.timeoutMs = timeoutMs;
        this.bean = bean;
        this.method = method;
        Class<?> originClass = ReflectUtils.getSpringProxyOriginClass(bean.getClass());
        this.location = originClass.getName() + "#" + method.getName();
        this.pidAndHost = ManagementFactory.getRuntimeMXBean().getName();
    }

    @Override
    public void run() {
        // 如果服务器还没启动完成，先不执行定时任务，防止某些资源没初始化好，一运行就出错了
        if (null == SpringContext.getStartedTime()) {
            return;
        }
        // 服务器重启时会导致正在执行的定时任务中断
        // 这里通过shutdownHook增加定时任务暂停状态，服务关闭的时候停止定时任务执行
        if (CronInitializer.SHUTDOWN_FLAG) {
            log.info("Run {}-Task ShuttingDown: {}", runMode, location);
            return;
        }
        // 如果发现前一个任务还没执行完，日志打印之后本次就不再执行，防止出现任务堆积爆内存（但定时检查不会停止）
        if (running) {
            log.error("Run {}-Task LastTaskStillRunning: runningMs={}, location={}", runMode, System.currentTimeMillis() - lastStartTime, location);
            return;
        }

        lastStartTime = System.currentTimeMillis();
        running = true;
        boolean actuallyRun = false;
        log.debug("Run {}-Task START: {}", runMode, location);
        try {
            switch (runMode) {
                case Everyone: // 这种模式下，不需要【决定谁来执行】，每个实例都直接执行任务即可
                    actuallyRun = true;
                    invoke();
                    break;
                case ChampionOnly: // 这种模式下，需要通过【抢令牌】的方式决定由哪个实例来执行任务

                    // 抢令牌流程（通过redis的setnx命令完成）：
                    //
                    // 1、抢占key1 = 任务位置 + 任务开始时间戳
                    //    由于具体任务是由CronExecutorUtils派发执行的，里面提供了一个方法getThisTimestamp()返回本次任务的开始时间
                    //    这个任务开始时间就可以作为本次任务的标记，当各实例的系统时间有差异时，只要差距不超过redisLockExpireMs就可以防止重复执行
                    //
                    //    举例：实例A系统时间17:29:10，实例B系统时间17:30:20，如果定时任务是每1min执行一次
                    //          如果没有抢占key1的机制，则17:31的任务，A和B都会各自再执行一遍
                    //          有key1的机制时，B会先抢占执行，A不会执行17:31的任务
                    //
                    // 2、在抢到key1后，再 抢占key2 = 任务位置
                    //    抢到后就可以直接开始执行任务，并由【定时任务超时扫描】任务（位于CronInitializer）扫描并不断续期key2
                    //    执行完毕后将key2删除
                    //    这个机制是为了处理任务执行耗时比较长的定时任务，防止在一个实例还在执行时，另一个实例到了时间启动新任务
                    //
                    //    举例：定时任务每1min执行一次，但执行耗时不定，某些时候会执行超过1min
                    //          现在有A和B两个实例，系统时间已做校正，在17:30时开始定时任务，B抢到了17:30的key1，并在接下来抢到了key2，开始执行任务
                    //          任务执行超时，在17:31时仍未结束，A开始做17:31的新任务检查，由于B还在执行中，并不会去抢17:31的key1
                    //          17:31的key1会被A抢到，但由于B一直在给key2续期，A抢不到key2故不会进入执行
                    //          如果任务继续超时，由于有以上的机制，其他实例抢不到令牌，不会新开任务
                    //          任务执行完后，key2被删除，其他实例可以在下一次任务开始时间参与抢令牌

                    Boolean key1obtained = false;
                    Boolean key2obtained = false;
                    CronTaskMini cronTaskMini = new CronTaskMini(pidAndHost, location, cron, runMode, timeoutMs);
                    String cronTaskJson = JacksonUtils.toJson(cronTaskMini);
                    RedisConnectionFactory redisConnectionFactory = customizer.getRedisConnectionFactory();
                    try (RedisConnection conn = redisConnectionFactory.getConnection()) {
                        byte[] key1 = (REDIS_LOCK_PREFIX + location + CronExecutorUtils.getThisTimestamp()).getBytes(StandardCharsets.UTF_8);
                        if (null != cronTaskJson) {
                            byte[] value = cronTaskJson.getBytes(StandardCharsets.UTF_8);
                            key1obtained = conn.set(key1, value, Expiration.from(REDIS_LOCK_KEY1_EXPIRE_MS, TimeUnit.MILLISECONDS), SetOption.SET_IF_ABSENT);
                        }
                    }
                    if (null != key1obtained && key1obtained) {
                        try (RedisConnection conn = redisConnectionFactory.getConnection()) {
                            byte[] key2 = (REDIS_LOCK_PREFIX + location).getBytes(StandardCharsets.UTF_8);
                            byte[] value = cronTaskJson.getBytes(StandardCharsets.UTF_8);
                            key2obtained = conn.set(key2, value, Expiration.from(REDIS_LOCK_KEY2_EXPIRE_MS, TimeUnit.MILLISECONDS), SetOption.SET_IF_ABSENT);
                        }
                        if (null != key2obtained && key2obtained) {
                            log.debug("Run {}-Task ObtainLockSuccess: {}", runMode, location);
                            obtainLock = true;
                            actuallyRun = true;
                            invoke();
                            break;
                        }
                    }
                    log.debug("Run {}-Task END-ObtainLockFailed(key1={}|key2={}): {}", runMode, key1obtained, key2obtained, location);
                    // 查询任务信息，如果发现redis里面存的任务配置和本机的不一致，需要报错提醒
                    try (RedisConnection conn = redisConnectionFactory.getConnection()) {
                        byte[] key2 = (REDIS_LOCK_PREFIX + location).getBytes(StandardCharsets.UTF_8);
                        CronTaskMini mini = JacksonUtils.toBean(conn.get(key2), CronTaskMini.class);
                        if (null != mini) {
                            StringBuilder sb = new StringBuilder();
                            if (!Objects.equals(mini.cron, cron) || !Objects.equals(mini.runMode, runMode)) {
                                sb.append("====================\n");
                                sb.append("config from ").append(mini.pidAndHost).append("\n");
                                sb.append("cron=").append(mini.cron).append("\n");
                                sb.append("runMode=").append(mini.runMode).append("\n");
                                sb.append("====================\n");
                                sb.append("config from ").append(pidAndHost).append("\n");
                                sb.append("cron=").append(cron).append("\n");
                                sb.append("runMode=").append(runMode).append("\n");
                                throw new IllegalArgumentException("Conflict Config: \nlocation=" + location + "\n\n" + sb);
                            }
                        }
                    }
                    break;
                default:
                    throw new IllegalArgumentException("Unsupported run mode: " + runMode);
            }
        } catch (Throwable ex) {
            if (log.isErrorEnabled()) {
                log.error("{}", StackTraceUtils.printThrowableSimple(ex), ex);
            }
            Map<String, String> mdc = MDC.getCopyOfContextMap();
            CronInitializer.cronExecutor.execute(() -> {
                try {
                    MDC.setContextMap(mdc);
                    customizer.onError(location, ex);
                } finally {
                    MDC.clear();
                }
            });
        } finally {
            lastCostMs = System.currentTimeMillis() - lastStartTime;
            running = false;
            if (obtainLock) {
                RedisConnectionFactory redisConnectionFactory = customizer.getRedisConnectionFactory();
                try (RedisConnection conn = redisConnectionFactory.getConnection()) {
                    byte[] key2 = (REDIS_LOCK_PREFIX + location).getBytes(StandardCharsets.UTF_8);
                    conn.del(key2);
                }
            }
            obtainLock = false;
            if (actuallyRun) {
                log.debug("Run {}-Task END in {}ms: {}", runMode, System.currentTimeMillis() - lastStartTime, location);
            }
        }
    }

    void renewLockIfObtained() {
        if (obtainLock && !CronInitializer.SHUTDOWN_FLAG) {
            RedisConnectionFactory redisConnectionFactory = customizer.getRedisConnectionFactory();
            try (RedisConnection conn = redisConnectionFactory.getConnection()) {
                byte[] key2 = (REDIS_LOCK_PREFIX + location).getBytes(StandardCharsets.UTF_8);
                conn.expire(key2, REDIS_LOCK_KEY2_EXPIRE_MS);
            }
        }
    }

    /**
     * 执行原始方法
     */
    void invoke() throws Throwable {
        try {
            this.method.invoke(this.bean);
        } catch (InvocationTargetException ex) {
            // 如果是执行方法内容时抛出异常，需要【提取到原始的异常】然后再抛出，这样报错提示才会更清晰
            if (null != ex.getCause()) {
                throw ex.getCause();
            }
            throw ex;
        }
    }

    public String getLocation() {
        return location;
    }

    public String getCron() {
        return cron;
    }

    public CronRunModeEnum getRunMode() {
        return runMode;
    }

    public long getTimeoutMs() {
        return timeoutMs;
    }

    public long getLastStartTime() {
        return lastStartTime;
    }

    public boolean isRunning() {
        return running;
    }

    public long getLastCostMs() {
        return lastCostMs;
    }

    @Override
    public int hashCode() {
        return location.hashCode();
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        CronTask that = (CronTask) o;
        return location.equals(that.location);
    }

    static class CronTaskMini {

        private String pidAndHost;

        private String location;

        private String cron;

        private CronRunModeEnum runMode;

        private long timeoutMs;

        public CronTaskMini() {
        }

        public CronTaskMini(String pidAndHost, String location, String cron, CronRunModeEnum runMode, long timeoutMs) {
            this.pidAndHost = pidAndHost;
            this.location = location;
            this.cron = cron;
            this.runMode = runMode;
            this.timeoutMs = timeoutMs;
        }

        public String getPidAndHost() {
            return pidAndHost;
        }

        public void setPidAndHost(String pidAndHost) {
            this.pidAndHost = pidAndHost;
        }

        public String getLocation() {
            return location;
        }

        public void setLocation(String location) {
            this.location = location;
        }

        public String getCron() {
            return cron;
        }

        public void setCron(String cron) {
            this.cron = cron;
        }

        public CronRunModeEnum getRunMode() {
            return runMode;
        }

        public void setRunMode(CronRunModeEnum runMode) {
            this.runMode = runMode;
        }

        public long getTimeoutMs() {
            return timeoutMs;
        }

        public void setTimeoutMs(long timeoutMs) {
            this.timeoutMs = timeoutMs;
        }
    }
}
